package com.tcm.kafka;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.config.TopicConfig;

import java.util.*;
import java.util.concurrent.ExecutionException;

public class KafkaAdminClientDemo {

    static String brokeList = "localhost:9092";
    static String topic = "topic-admin";

    public static void createTopic(){
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokeList);
        properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        AdminClient client = KafkaAdminClient.create(properties);

//        Map<Integer, List<Integer>> replicasAssignments = new HashMap();
//        replicasAssignments.put(0, Arrays.asList(0));
//        replicasAssignments.put(1, Arrays.asList(0));
//        replicasAssignments.put(2, Arrays.asList(0));
//        replicasAssignments.put(3, Arrays.asList(0));
//        NewTopic newTopic = new NewTopic(topic, replicasAssignments);

        NewTopic newTopic = new NewTopic(topic, 4, (short)1);

//        Map<String, String> configs = new HashMap();
//        configs.put("cleanup.policy", "compact");
//        newTopic.configs(configs);

        CreateTopicsResult result = client.createTopics(Collections.singletonList(newTopic));

        try{
            result.all().get();
        }catch (Exception e){
            e.printStackTrace();
        }
        client.close();
    }

    public static void describeTopicConfig() throws ExecutionException, InterruptedException {
        String brokerList = "localhost:9092";
        String topic = "topic-admin";

        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        AdminClient client = AdminClient.create(props);

        ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
        DescribeConfigsResult result = client.describeConfigs(Collections.singleton(resource));
        Config config = result.all().get().get(resource);
        System.out.println(config);
        client.close();
    }

    public static void alterTopicConfig() throws ExecutionException, InterruptedException {
        String brokeList = "localhost:9092";
        String topic = "topic-admin";
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokeList);
        props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        AdminClient client = AdminClient.create(props);

        ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
        ConfigEntry entry = new ConfigEntry("cleanup.policy", "compact");
        Config config = new Config(Collections.singleton(entry));
        Map<ConfigResource, Config> configs = new HashMap();
        configs.put(resource, config);
        AlterConfigsResult result = client.alterConfigs(configs);
        result.all().get();
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        createTopic();
//        describeTopicConfig();
        alterTopicConfig();
    }
}
